package com.erainm.customserializer;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;
import java.util.Map;

/**
 * Created by erainm on 2021/9/9 4:35 下午.
 *
 * @Description TODO:
 **/
public class Myproducer {
    public static void main(String[] args) {
        Map<String, Object> configs = new HashMap<>();
        // 初始连接用到的broker地址
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "lg04:9092");
        // 指定key的序列化类
        configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        // 指定value的序列化类
        configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, UserSerializer.class);
        KafkaProducer<String, User> producer = new KafkaProducer<>(configs);

//        User user = new User(1001,"张三");
//        User user = new User(1001,"李四");
//        User user = new User(1001,"王五");
        User user = new User(1001,"赵六");


        ProducerRecord<String,User> record = new ProducerRecord<String,User>("tp_user_01",user.getUsername(),user);
        producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (exception!=null){
                    System.out.println("消息发送异常");
                }else {
                    System.out.println("主题：" + metadata.topic() + "\t"
                    + "分区：" + metadata.partition() + "\t"
                    + "生产者偏移量：" + metadata.offset());
                }
            }
        });
        producer.close();
    }
}
